package chapter03
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
object Test26_avgSalary {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    val conf = new SparkConf().setMaster("local[*]").setAppName("avgSalary")
    val sc = new SparkContext(conf)
    val value = sc.textFile("input/Employee_salary_first_half.csv")
    val value1 = sc.textFile("input/Employee_salary_second_half.csv")
    //去掉表头
    val value2 = value.mapPartitionsWithIndex((index, e) => {
      if (index == 0) e.drop(1) else e
    })
    val value3 = value1.mapPartitionsWithIndex((index, e) => {
      if (index == 0) e.drop(1) else e
    })
    //取得每个数据集中的姓名和工资列
    val value4 = value2.map(e => e.split(","))
      .map(e => (e(1), e(5).toInt))
    val value5 = value3.map(e => e.split(","))
      .map(e => (e(1), e(5).toInt))
    //合并数据集
    val value6 = value4.union(value5)
    //使用combineBykey (string,sum,count)
    val value7 = value6.combineByKey(
      v => (v, 1),
      (acc: (Int, Int), v) => (acc._1 + v, acc._2 + 1),
      (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
    )
    val tuples = value7
      .map(e => (e._1, e._2._1.toDouble / e._2._2))
      .sortBy(e => e._2,false)
      .take(3)
    println(tuples.toList)
    sc.stop()
  }
}
